#pragma once
#include "rpc_caller.hpp"
#include "rpc_requestor.hpp"
#include "../common/net.hpp"
#include "../common/dispatcher.hpp"
#include "rpc_register.hpp"
#include <future>
#include <functional>

namespace RPC
{
    namespace Client
    {
        class TopicManager
        {
        public:
            using Ptr = std::shared_ptr<TopicManager>;
            using MsgCallBack = std::function<void(const std::string &type, const std::string &msg)>;

            TopicManager(const Requestor::Ptr& req) :_requestor(req){}
            bool topicSubscribe(const BaseConnection::Ptr &conn, const std::string &topic, const MsgCallBack &cb)
            {
                insert(topic, cb);
                bool ret = commonRequest(conn, topic, TopicOpType::TOPIC_SUB);
                if(ret == false)
                {
                    remove(topic);
                    return false;
                }
                return true;
            }
            bool topicCancle(const BaseConnection::Ptr &conn, const std::string &topic)
            {
                remove(topic);
                return commonRequest(conn, topic, TopicOpType::TOPIC_UNSUB);
            }
            bool topicCreate(const BaseConnection::Ptr &conn, const std::string &topic)
            {
                return commonRequest(conn, topic, TopicOpType::TOPIC_CREAT);
            }
            bool topicRemove(const BaseConnection::Ptr &conn, const std::string &topic)
            {
                return commonRequest(conn, topic, TopicOpType::TOPIC_DELETE);
            }
            bool topicPublish(const BaseConnection::Ptr &conn, const std::string &topic, const std::string &msg)
            {
                return commonRequest(conn, topic, TopicOpType::TOPIC_PUBLISH, msg);
            }

            void onPublish(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &tprequest)
            {
                if(tprequest->opType() != TopicOpType::TOPIC_PUBLISH) return;
                const MsgCallBack& cb = find(tprequest->topicKey());
                if(cb == nullptr)
                {
                    ERROR_LOG("找不到对应的回调函数！");
                    return;
                }
                return cb(tprequest->topicKey(), tprequest->message());
            }

        private:
            bool commonRequest(const BaseConnection::Ptr &conn, const std::string &topic,
                               const TopicOpType &type, const std::string &msg = "")
            {
                TopicRequest::Ptr tprequest = MessageFactory::create<TopicRequest>();
                tprequest->setId(UUID::uuid());
                tprequest->setOpType(type);
                tprequest->setTopicKey(topic);
                tprequest->setType(MType::REQ_TOPIC);
                if (type == TopicOpType::TOPIC_PUBLISH)
                {
                    tprequest->setMessage(msg);
                }
                BaseMessage::Ptr resp;
                bool sendret = _requestor->send(conn, tprequest, resp);
                if (sendret == false)
                {
                    ERROR_LOG("消息发送失败！");
                    return false;
                }
                TopicResponse::Ptr tpresponse = std::dynamic_pointer_cast<TopicResponse>(resp);
                if (tprequest.get() == nullptr)
                {
                    ERROR_LOG("向下转型失败！");
                    return false;
                    ;
                }
                if (tpresponse->rCode() != RCode::RCODE_OK)
                {
                    ERROR_LOG("消息返回值错误！");
                    return false;
                    ;
                }
                return true;
            }

            void insert(const std::string &topic, const MsgCallBack &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callback.insert({topic, cb});
            }

            void remove(const std::string& topic)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callback.erase(topic);
            }

            const MsgCallBack find(const std::string &topic)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callback.find(topic);
                if (it == _topic_callback.end())
                {
                    return MsgCallBack();
                }
                return it->second;
            }

            std::mutex _mutex;
            std::unordered_map<std::string, MsgCallBack> _topic_callback; // 收到消息调用对应的回调函数
            Requestor::Ptr _requestor;                                    // 保证收到消息时一一对应
        };
    }
}